package com.lefu.risk.storm.topology;

import org.apache.storm.Config;
import org.apache.storm.LocalCluster;
import org.apache.storm.generated.StormTopology;
import org.apache.storm.topology.TopologyBuilder;

import com.lefu.risk.storm.bolt.Abolt;
import com.lefu.risk.storm.content.Content;
import com.lefu.risk.storm.spout.TestSpout;

public class TopoX {
	public static void main(String[] args) throws Exception {
		TopologyBuilder builder = new TopologyBuilder();
		
		TestSpout tsp = new TestSpout();
		Abolt ba = new Abolt();
		
		builder.setSpout(Content.SENTENCE_SPOUT_ID,tsp,3);
		builder.setBolt(Content.SPLIT_BOLT_ID,ba,3).shuffleGrouping(Content.SENTENCE_SPOUT_ID);
		
		StormTopology topology = builder.createTopology();
		
		Config config = new Config();
		
		LocalCluster cluster = new LocalCluster();
		cluster.submitTopology(Content.WORD_COUNT_TOPOLOGY_NAME, config, topology);
		
		Thread.sleep(15000);
		
		cluster.killTopology(Content.WORD_COUNT_TOPOLOGY_NAME);
		cluster.shutdown();
		
//		StormSubmitter.submitTopology(Content.WORD_COUNT_TOPOLOGY_NAME, config, topology);
	}
}
